#include <ev.h>
#include <amqpcpp.h>
#include <amqpcpp/libev.h>
#include <openssl/ssl.h>
#include <openssl/opensslv.h>
#include "logger.hpp"

class MQClient {
public:
    using MessageCallback = std::function<void(const char*, size_t sz)>;

    MQClient(const std::string &user, 
        const std::string &passwd, 
        const std::string &host)
    {
        _loop = EV_DEFAULT;
        _handler = std::make_unique<AMQP::LibEvHandler>(_loop);
        std::string url = "amqp://" + user + ":" + passwd + "@" + host + "/";
        AMQP::Address address(url);
        _connection = std::make_unique<AMQP::TcpConnection>(_handler.get(), address);
        _channel = std::make_unique<AMQP::TcpChannel>(_connection.get());

        _loop_thread = std::thread([this]() {
            ev_run(_loop, 0);
        });
    }

    ~MQClient()
    {
        struct ev_async async_watcher;
        ev_async_init(&async_watcher, watcher_callabck);
        ev_async_start(_loop, &async_watcher);
        ev_async_send(_loop, &async_watcher);
        _loop_thread.join();
        _loop = nullptr;
    }

    void declareComponents(const std::string &exchange,
        const std::string &queue,
        const std::string &routing_key = "routing_key", 
        AMQP::ExchangeType exchange_type = AMQP::ExchangeType::direct)
    {
        // 5.声明交换机
        _channel->declareExchange(exchange, exchange_type)
        .onError([](const char* message) {
            LOG_ERROR("声明交换机失败: {}", message);
            exit(0); 
        })
        .onSuccess([exchange]() {
            LOG_INFO("{}交换机创建成功!", exchange);
        });
        // 6.声明队列
        _channel->declareQueue(queue)
        .onError([](const char* message) {
            LOG_ERROR("声明队列失败: {}", message);
            exit(0);
        })
        .onSuccess([queue]() {
            LOG_INFO("{}队列创建成功!", queue);
        });
        // 7.绑定队列到交换机
        _channel->bindQueue(exchange, queue, routing_key)
        .onError([exchange, queue](const char* message) {
            LOG_ERROR("{}-{}绑定失败: {}", exchange, queue, message);
            exit(0);
        })
        .onSuccess([exchange, queue]() {
            LOG_INFO("{}-{}绑定成功!", exchange, queue);
        });
    }

    bool publish(const std::string &exchange,
        const std::string &message, 
        const std::string &routing_key = "routing_key")
    {
        bool ret = _channel->publish(exchange, routing_key, message);
        if(ret) {
            LOG_INFO("消息发布成功: {}", message);
            return true;
        } else {
            LOG_ERROR("消息发布失败: {}", message);
            return false;
        }
    }

    void consume(const std::string &queue, const MessageCallback &callback)
    {
        _channel->consume(queue, "consume_tag")
        .onReceived([this, callback](const AMQP::Message &message, 
            uint64_t deliveryTag, 
            bool redelivered) {
            callback(message.body(), message.bodySize());
            _channel->ack(deliveryTag);
            std::string msg(message.body(), message.bodySize());
            LOG_INFO("收到消息: {}", msg);
        })
        .onError([queue](const char* message) {
            LOG_ERROR("订阅{}队列消息失败: {}", queue, message);
            exit(0);
        });
    }

private:
    static void watcher_callabck(struct ev_loop *loop, ev_async *watcher, int32_t revents) 
    {
        ev_break(loop, EVBREAK_ALL);
    }

private:
    struct ev_loop *_loop;
    std::unique_ptr<AMQP::LibEvHandler> _handler;
    std::unique_ptr<AMQP::TcpConnection> _connection;
    std::unique_ptr<AMQP::TcpChannel> _channel;
    std::thread _loop_thread;
};